package com.ht.mq;

import com.ht.controller.SendController;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component // 注入spring容器
// topic和消费者发送的topic相同    group 不用和生产者group相同
@RocketMQMessageListener(topic = "ht-code-topic",consumerGroup = "ht-code-topic")
public class SendMassageMQ implements RocketMQListener, RocketMQPushConsumerLifecycleListener{

    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private SendController sendController;

    @Override
    public void onMessage(Object phoneNum) {

    }
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    System.out.println("开始消费");
                    String phoneNum = "";
                    String code = "";
                    //进行遍历 消息
                    for (MessageExt messageExt: list) {
                        //获取到的消息内容是 byte[]格式  转成String类型 (就是整个消息体)
                       phoneNum = new String(messageExt.getBody(),"UTF-8");
                    //调用阿里云 短信业务 进行发送短信  返回值是验证码
                        code = sendController.sendMassage(phoneNum);
                    }
                    //将 手机号作为key 验证码 作为value 放到Redis里面 进行存储
                    redisTemplate.boundValueOps(phoneNum).set(code);
                    //设置超时时间为  1  天
                    redisTemplate.expire(phoneNum,1, TimeUnit.DAYS);
                    System.out.println("消费成功");
                }catch (Exception e){
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });


    }
}
